package com.ztesoft.huangfu.learning.springboot.es.service.impl;

import com.ztesoft.huangfu.learning.springboot.es.repository.StudentDocRepository;
import com.ztesoft.huangfu.learning.springboot.es.service.EsAggregationsSearch;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.InternalDateHistogram;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.cardinality.CardinalityAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.min.MinAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.tophits.TopHits;
import org.elasticsearch.search.aggregations.metrics.valuecount.InternalValueCount;
import org.elasticsearch.search.aggregations.metrics.valuecount.ValueCountAggregationBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;

/**
 * Created with IntelliJ IDEA.
 * Author: huangfu.daohui
 * Date: 2018/12/20
 * Time: 16:13
 * Version: 1.0
 * Description:Es聚合查询
 **/
@Service
public class EsAggregationsSearchImpl implements EsAggregationsSearch {

    private Logger log = LoggerFactory.getLogger(getClass());

    @Autowired
    private StudentDocRepository studentDocRepository;

    @Autowired
    ElasticsearchTemplate elasticsearchTemplate;

    /**
     * 计算：求和、最大、最小、平均等
     *
     * @return
     */
    @Override
    public String calculation() {

        // 设置查询条件
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("age").from("20").lte("30"));
        // 聚合查询，根据条件过滤的出来的结果对age求和
        SumAggregationBuilder sumBuilder = AggregationBuilders.sum("sum_age").field("age");//sum

        //根据条件过滤的出来的结果对age求取平均值
        AvgAggregationBuilder avgBuilder = AggregationBuilders.avg("age_age").field("age");//sum

        //根据条件过滤的出来的结果对age取最小值
        MinAggregationBuilder minBuilder = AggregationBuilders.min("min_age").field("age");//sum

        //根据条件过滤的出来的结果对age去最大值
        MaxAggregationBuilder maxBuilder = AggregationBuilders.max("max_age").field("age");//sum

        // 据条件过滤的出来的结果对age统计数量
        ValueCountAggregationBuilder vcab = AggregationBuilders.count("count_age").field("uid");

        // 据条件过滤的出来的结果对name去除重名
        CardinalityAggregationBuilder cb = AggregationBuilders.cardinality("distinct_count_name").field("name");

        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder)
                .addAggregation(sumBuilder)
                .addAggregation(vcab) // 多聚合在response里面分别获取,两个完全独立的聚合
                .build();
        String saleAmount = elasticsearchTemplate.query(searchQuery, response -> {
            InternalSum sum = (InternalSum) response.getAggregations().asList().get(1);
            InternalValueCount count = (InternalValueCount) response.getAggregations().asList().get(0);
            //return sum.getValue( );
            return "sum:=" + sum.getValue() + ",count=" + count.getValue();
        });
        return saleAmount;
    }

    /**
     * 分组
     *
     * @return
     */
    @Override
    public String term() {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("age").from("20").lte("30"));
        // 聚合查询，根据年龄，分组，然后求和
        TermsAggregationBuilder sumBuilder = AggregationBuilders.terms("term_age")
                .field("age")
                .order(BucketOrder.count(true));//先分组，然后按照各组的数量排序，适合top10

        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder)
                .addAggregation(sumBuilder)
                .build();
        String result = elasticsearchTemplate.query(searchQuery, response -> {

            // 两种方式获取聚合的结果：Terms和TopHits
            Terms terms = response.getAggregations().get("term_age");
            TopHits topHits = response.getAggregations().get("term_age");

            String info = "";
            for (Terms.Bucket bucket : terms.getBuckets()) {
                String key = bucket.getKeyAsString();//每组的key，就是beginTime
                info += "\r\n groupName=" + key + "  count =" + bucket.getDocCount();//每组的数量
            }
            return info;
        });
        return result;
    }

    /**
     * 根据时间分组
     * 日期直方图聚合
     * /* {
     * "aggs":{
     * "articles_over_time":{
     * "date_histogram":{
     * "field":"recordTime",
     * "interval":"day",
     * "format":"yyyy-MM-dd"
     * }
     * }
     * }
     * }
     *
     * @return
     */
    @Override
    public String date_histogram() {
        // 设置查询条件，根据查询结果分组
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("age").from(20).lte(30));
        // 聚合查询。goodsSales是要统计的字段，sum_sales是自定义的别名
        //分组。。但是文本里有空格，冒号这种会被分隔为不同的组。。比如20180905T11:19:20 会被分隔为四组：20180905T11、11、19、20

        AbstractAggregationBuilder statistic = AggregationBuilders.dateHistogram("statistic").field("birthday")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd");//分组

        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder)
                .addAggregation(statistic)
                //.addAggregation(countBuilder)
                .build();

        String result = elasticsearchTemplate.query(searchQuery, response -> {
            InternalDateHistogram terms = response.getAggregations().get("statistic");
            String info = "";
            for (InternalDateHistogram.Bucket bucket : terms.getBuckets()) {
                String key = bucket.getKeyAsString();//每组的key，就是beginTime
                info += "\r\n groupName=" + key + "  count =" + bucket.getDocCount();//每组的数量
            }
            return info;
        });
        return result;
    }

    /**
     * 子聚合查询
     *
     * @return
     */
    @Override
    public String subAggregation() {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("age").from("20").lte("30"));
        // 聚合查询，根据年龄，分组，年龄一样的分为一组
        TermsAggregationBuilder sumBuilder = AggregationBuilders.terms("term_age").field("age");//sum

        // 作为子聚合，根据年龄分组成多个bucket之后，每个bucket再根据sex分组
        TermsAggregationBuilder builder2 = AggregationBuilders.terms("term_sex").field("sex");
        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(queryBuilder)
                .addAggregation(sumBuilder.subAggregation(builder2))// 可增加聚合后排序
                .build();
        String result = elasticsearchTemplate.query(searchQuery, response -> {
            Terms termsAge = response.getAggregations().get("term_age");
            String info = "";
            // 二层循环取值
            for (Terms.Bucket bucket : termsAge.getBuckets()) {
                String key = bucket.getKeyAsString();//每组的key，就是beginTime
                info += "\r\n groupName=" + key + "  count =" + bucket.getDocCount();//每组的数量
                Terms termSex = bucket.getAggregations().get("term_sex");

                for (Terms.Bucket sex : termSex.getBuckets()) {
                    String keySex = bucket.getKeyAsString();//每组的key，就是beginTime
                    System.out.println("key:" + keySex + "；countSex:" + sex.getDocCount());
                }
            }
            return info;
        });
        return result;
    }
}
